package cn.zwx.common.rocket.producer.service.impl;

import cn.zwx.base.enums.ResponseCodeEnum;
import cn.zwx.base.exception.BizWrapException;
import cn.zwx.common.rocket.producer.bo.MessageBO;
import cn.zwx.common.rocket.producer.bo.TransactionListenerBO;
import cn.zwx.common.rocket.producer.constants.RocketProducerConstants;
import cn.zwx.common.rocket.producer.listener.DefaultTransactionListener;
import cn.zwx.common.rocket.producer.service.TransactionMqProducerService;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

/**
 * @author zhangwenxue
 */
@Service
public class TransactionMqProducerServiceImpl  implements TransactionMqProducerService {

    public static final Logger logger = LoggerFactory.getLogger(TransactionMqProducerServiceImpl.class);
    @Resource
    private TransactionMQProducer transactionProducer;

    @Override
    public void start() {
        // 在启动之前设置回查
        List<TransactionListenerBO> checkListenerBOList = RocketProducerConstants.TRANSACTION_LISTENER_LIST;
        if (!CollectionUtils.isEmpty(checkListenerBOList)){
            transactionProducer.setTransactionListener(new DefaultTransactionListener(checkListenerBOList));
        }
    }

    @Override
    public void shutDown() {
        if (Objects.nonNull(transactionProducer)){
            transactionProducer.shutdown();
        }
    }

    @Override
    public SendResult sendMsg(MessageBO bo, Object args) throws BizWrapException {
        Message message = new Message(bo.getTopic(), bo.getTag(), bo.getKey(), bo.getMessageBody().getBytes());
        SendResult sendResult;
        try {
            sendResult = transactionProducer.sendMessageInTransaction(message,args);
        } catch (MQClientException e) {
            logger.error("Send mq message failed. Topic is:{} Tag:{} key:{},body:{}",message.getTopic(), message.getTags(),message.getKeys(),
                    new String(message.getBody()),e);
            throw new BizWrapException(ResponseCodeEnum.SEND_MQ_ERROR);
        }
        logger.info("Send mq message success. Topic is:{} Tag:{} key:{} body:{}",message.getTopic(), message.getTags(),
                message.getKeys(),new String(message.getBody()));
        return sendResult;
    }
}
